package com.liuwei.spring.cloud.configuration;

import com.liuwei.spring.cloud.biz.support.GlobalExceptionUtils;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.task.TaskDecorator;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author : ThreadPoolConfiguration
 * @version V1.0
 * @date: 2023/12/22 10:00 AM
 * @Description: TODO
 */
@Slf4j
@Configuration
public class ThreadPoolConfiguration {

    public void executors(){
        ExecutorService service = Executors.newFixedThreadPool(5);
    }

    /**
     * ThreadPoolTaskExecutor instanceof AsyncTaskExecutor，Executor
     * Sleuth 默认会处理 ThreadPoolTaskExecutor，ExecutorService，Executor，AsyncTaskExecutor 等
     * @return
     */
    @Primary
    @Bean("CompletableFuturePool")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        //
        ThreadPoolTaskExecutor executor = new SpringMdcThreadPoolWrapper();
        // 核心线程数：线程池创建时候初始化的线程数
        executor.setCorePoolSize(10);
        // 最大线程数：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
        executor.setMaxPoolSize(20);
        // 缓冲队列：用来缓冲执行任务的队列
        executor.setQueueCapacity(200);
        // 允许线程的空闲时间60秒：当超过了核心线程之外的线程在空闲时间到达之后会被销毁
        executor.setKeepAliveSeconds(60);
        // 线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池
        executor.setThreadNamePrefix("CompletableFuturePool-");
        // 缓冲队列满了之后的拒绝策略：由调用线程处理（一般是主线程）
        //AbortPolicy：丢弃任务并抛出 RejectedExecutionException 异常
        //DiscardPolicy：丢弃任务，但是不抛出异常。可能导致无法发现系统的异常状态
        //DiscardOldestPolicy：丢弃队列最前面的任务，然后重新提交被拒绝的任务
        //CallerRunsPolicy：不丢弃任务 由调用线程处理该任务
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 线程池中corePoolSize线程空闲时间达到keepAliveTime也将关闭
        executor.setAllowCoreThreadTimeOut(true);
        // 线程池统一异常处理
        executor.setTaskDecorator(new TaskDecorator() {
            @Override
            public Runnable decorate(Runnable runnable) {
                return () -> {
                    try {
                        runnable.run();
                    } catch (Exception e) {
                        // 处理异常，例如记录日志
                        log.error("Async task failed", e);
                        throw e; // 重新抛出异常以便调用者知道发生了错误
                    }
                };
            }
        });
        executor.initialize();
        return executor;
    }

    /**
     * ThreadPoolExecutor instanceof ExecutorService，Executor
     * @return
     */
    @Bean("FeishuPool")
    public ThreadPoolExecutor threadPoolExecutor() {
        //获取cpu核心数
        final int cpu = Runtime.getRuntime().availableProcessors();
        //设置核心线程池数量
        final int corePoolSize = cpu + 1;
        //设置线程池中最大线程数量
        final int maximumPoolSize = cpu * 2 + 1;
        //设置线程空闲时间，超时终止线程，线程池线程数量维持在核心线程池大小(corePoolSize)
        final long keepAliveTime = 1L;
        //设置空闲时间的时间单位
        final TimeUnit timeUnit = TimeUnit.SECONDS;
        //设置阻塞队列，用来存储等待执行的任务，如果当前线程数量已超过核心线程设定的数量，则把任务暂时存放在等待队列中
        final int maxQueueNum = 1 << 7;
        //new CustomThreadFactory()：线程工程，用来创建线程
        //new ThreadPoolExecutor.AbortPolicy()：如果线程池已满，新的任务的处理方式
        ThreadPoolExecutor executor = new JdkThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                timeUnit,
                new LinkedBlockingQueue<Runnable>(maxQueueNum),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );
        // ThreadPoolExecutor 无法自定义 TaskDecorator 方法
        return executor;
    }

    /**
     * ThreadPoolExecutor是JDK中的JUC
     */
    @Slf4j
    public static class JdkThreadPoolExecutor extends ThreadPoolExecutor {

        public JdkThreadPoolExecutor(int corePoolSize,
                                     int maximumPoolSize,
                                     long keepAliveTime,
                                     TimeUnit unit,
                                     BlockingQueue<Runnable> workQueue) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        }

        public JdkThreadPoolExecutor(int corePoolSize,
                                     int maximumPoolSize,
                                     long keepAliveTime,
                                     TimeUnit unit,
                                     BlockingQueue<Runnable> workQueue,
                                     RejectedExecutionHandler handler) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler);
        }

        public JdkThreadPoolExecutor(int corePoolSize,
                                     int maximumPoolSize,
                                     long keepAliveTime,
                                     TimeUnit timeUnit,
                                     LinkedBlockingQueue<Runnable> workQueue,
                                     ThreadFactory threadFactory,
                                     AbortPolicy abortPolicy) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, workQueue, threadFactory, abortPolicy);
        }

        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            log.error("CloudThreadPoolExecutor|error:{}", t.getMessage(),t);
            try {
                GlobalExceptionUtils.handleGlobalException(t);
            } catch (Exception exception) {
                log.error(t.getMessage());
            }
        }
    }

    /**
     * ThreadPoolTaskExecutor是spring core包中的
     */
    @Slf4j
    public static class SpringMdcThreadPoolWrapper extends ThreadPoolTaskExecutor {

        public SpringMdcThreadPoolWrapper() {

        }

        @Override
        public void execute(Runnable task) {
            super.execute(wrap(task, MDC.getCopyOfContextMap()));
        }

        @Override
        public void execute(Runnable task, long startTimeout) {
            super.execute(wrap(task, MDC.getCopyOfContextMap()), startTimeout);
        }

        @Override
        public <T> Future<T> submit(Callable<T> task) {
            return super.submit(wrap(task, MDC.getCopyOfContextMap()));
        }

        @Override
        public Future<?> submit(Runnable task) {
            return super.submit(wrap(task, MDC.getCopyOfContextMap()));
        }

        @Override
        public ListenableFuture<?> submitListenable(Runnable task) {
            return super.submitListenable(wrap(task, MDC.getCopyOfContextMap()));
        }
        @Override
        public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
            return super.submitListenable(wrap(task, MDC.getCopyOfContextMap()));
        }

        @Override
        public CompletableFuture<Void> submitCompletable(Runnable task) {
            return super.submitCompletable(wrap(task, MDC.getCopyOfContextMap()));
        }

        @Override
        public <T> CompletableFuture<T> submitCompletable(Callable<T> task) {
            return super.submitCompletable(wrap(task, MDC.getCopyOfContextMap()));
        }

        public static <T> Callable<T> wrap(final Callable<T> callable, final Map<String, String> context) {
            return () -> {
                if (context == null) {
                    MDC.clear();
                } else {
                    MDC.setContextMap(context);
                }
                return callable.call();
            };
        }

        public static Runnable wrap(final Runnable runnable, final Map<String, String> context) {
            return () -> {
                if (context == null) {
                    MDC.clear();
                } else {
                    MDC.setContextMap(context);
                }
                runnable.run();
            };
        }
    }

}
